package xyz.riceball.springcloudstreambinderredis.component;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RLock;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ObjectUtils;
import xyz.riceball.springcloudstreambinderredis.common.Constants;
import xyz.riceball.springcloudstreambinderredis.common.MessageDTO;

import java.time.LocalDateTime;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * RedisMessageProducerAdapter
 * </p>
 *
 * @author xiaovcloud
 * @since 2024/6/24 17:06
 */
public class RedisMessageProducerAdapter extends MessageProducerSupport {

	private final String name;
	private final RedissonClient redissonClient;
	final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);


	public RedisMessageProducerAdapter(String name, RedissonClient redissonClient) {
		this.name = name;
		this.redissonClient = redissonClient;
		scheduledExecutorService.scheduleAtFixedRate(() -> {
			RLock lock = redissonClient.getLock(name);
			try {
				lock.lock();
				RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(name);
				RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
				MessageDTO peek = (MessageDTO) delayedQueue.peek();
				if (peek==null){
					return;
				}
				LocalDateTime time = (LocalDateTime) peek.getHeaders().get(Constants.DELAY_MESSAGE_TIME);
				if (LocalDateTime.now().isAfter(time)) {
					MessageDTO poll = (MessageDTO) delayedQueue.poll();
					if (!ObjectUtils.isEmpty(poll)) {
						Message redisMessage = MessageBuilder.withPayload(poll.getPayload())
							.copyHeaders(poll.getHeaders())
							.build();
						this.sendMessage(redisMessage);
					}
				}
			} finally {
				if (lock.isLocked()) {
					lock.unlock();
				}
			}
		}, 0, 500, TimeUnit.MICROSECONDS);
	}

	@Override
	protected void onInit() {
		super.onInit();
		RTopic topic = redissonClient.getTopic(name);
		topic.addListener(MessageDTO.class, (channel, message) -> {
			Message redisMessage = MessageBuilder.withPayload(message.getPayload())
				.copyHeaders(message.getHeaders())
				.build();
			if (!redisMessage.getHeaders().containsKey(Constants.DELAY_MESSAGE)) {
				this.sendMessage(redisMessage);
			}
		});
	}

	/**
	 * Take no action by default.
	 * Subclasses may override this if they
	 * need lifecycle-managed behavior. Protected by 'lifecycleLock'.
	 */
	@Override
	protected void doStart() {
		super.doStart();
	}

	/**
	 * Take no action by default.
	 * Subclasses may override this if they
	 * need lifecycle-managed behavior.
	 */
	@Override
	protected void doStop() {
		super.doStop();
	}
}
